package com.nucarf.yc.app

import com.nucarf.yc.bean.{ChangeOrderLog, OrderInfo}
import com.nucarf.yc.sink.JDBCSink
import com.nucarf.yc.utils.{DTUtils, KafkaUtil}
import net.minidev.json.JSONObject
import net.minidev.json.parser.JSONParser
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

import java.util.UUID
import scala.collection.mutable

object ChangeOrder {


  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    var dml_name: String = ""

    /* 设置并行度 */
    env.setParallelism(1)

    /* 设置时间语义 */
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val inputStream: DataStream[String] =
      env.addSource( new FlinkKafkaConsumer[String](
        KafkaUtil.CONSUME_TEST,
        new SimpleStringSchema(),
        KafkaUtil.kafkaConsumerProperties()) )

    inputStream




//    val value: DataStream[(Double, Double)] = inputStream.map(new MyMapFunction)
//    inputStream.print()
//    value.print()
/*
    val logStream: DataStream[ChangeOrderLog] = inputStream.map(value => {

      val parser = new JSONParser(1)
      val nObject: JSONObject = parser.parse(value).asInstanceOf[JSONObject]
      val databaseName: String = nObject.get("database").toString
      val tableName: String = nObject.get("table").toString
      val ts: Int = nObject.get("ts").toString.toInt
      val operation: String = nObject.get("type").toString
      val jsonString: String = nObject.get("data").toString
      
      val p1 = new JSONParser(1)
      var id: String = ""
      if ("tb_order_all".equals(tableName)){
        id = p1.parse(nObject.get("data").toString).asInstanceOf[JSONObject].get("order_id").toString
      } else if("tb_nc_member_order_info".equals(tableName)){
        id = p1.parse(nObject.get("data").toString).asInstanceOf[JSONObject].get("moi_id").toString
      }
      ChangeOrderLog(id = id, ts = ts, tableName = tableName, databaseName = databaseName, operation = operation, createTime = DTUtils.nowString(), jsonString = jsonString)
    })

    logStream.print()
    logStream.addSink(new JDBCSink)*/

    val logStream: DataStream[ChangeOrderLog] = inputStream.map(value => {

      val parser = new JSONParser(1)
      val nObject: JSONObject = parser.parse(value).asInstanceOf[JSONObject]
      val databaseName: String = ""
      val tableName: String = ""
      val ts: Long = 22
      val operation: String = ""
      val jsonString: String = nObject.toString

      val p1 = new JSONParser(1)
      var id: String = ""

      ChangeOrderLog(id = id, System.currentTimeMillis(), tableName = tableName, databaseName = databaseName, operation = operation, createTime = DTUtils.nowString(), jsonString = jsonString)
    })

    inputStream.print()
//    logStream.addSink(new JDBCSink);

    env.execute("YC_ORDER_TEST : ")
  }

}

class MyMapFunction extends MapFunction[String, (Double, Double)]{

  var count: Double = 0
  var sum: Double = 0
  override def map(t: String): (Double, Double) = {
    val parser = new JSONParser(1)
    val nObject: JSONObject = parser.parse(t).asInstanceOf[JSONObject]
    val ts: Double = nObject.get("amount").toString.toDouble
    count = count + ts
    sum = sum + 1
    Tuple2(sum, count)
  }
}
